package cas.ihep.hdfsIn;

import cas.ihep.util.RedirectIO;
import cas.ihep.util.RefreshFileSize;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdfs.DFSClient;
import org.apache.hadoop.hdfs.Service4Hep;
import org.apache.hadoop.hdfs.protocol.HdfsFileStatus;
import org.apache.hadoop.hdfs.protocol.LocatedBlock;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.util.Shell;
import org.globus.ftp.*;
import org.globus.ftp.exception.ServerException;
import org.globus.ftp.extended.GridFTPServerFacade;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.lang.reflect.Field;
import java.net.URI;
import java.net.URISyntaxException;

/**
 * Created by chocolate on 2018/1/4.
 */
public class ApiMapper extends Mapper<LongWritable, Text, Text, Text> {
    DFSClient dfs = null;
    Service4Hep s4h = null;
    GridFTPClient client;
    long totalsize=0;
    long totaltime=0;

    public void run(Context context) throws IOException, InterruptedException{
        try {
            client = new GridFTPClient("helion02.ihep.ac.cn", 5555);
//            Class<?> klass = GridFTPClient.class;
//            Field gLocalServer_field = klass.getDeclaredField("gLocalServer");
//            gLocalServer_field.setAccessible(true);
//            GridFTPServerFacade gLocalServer = (GridFTPServerFacade) gLocalServer_field.get(client);
//            gLocalServer.setOptions(new RetrieveOptions(16));
            client.authorize("xingg", "123456");
            client.setType(GridFTPSession.TYPE_IMAGE);
            client.setMode(GridFTPSession.MODE_EBLOCK);
            client.setDataChannelAuthentication(DataChannelAuthentication.NONE);
            client.setDataChannelProtection(1);
            client.setOptions(new RetrieveOptions(3));
            super.run(context);
            client.close();
            System.out.println("Speed: "+(totalsize/(totaltime/1000.)/1024/1024)+"MB/s");
        }catch (ServerException e){
            throw new IOException(e);
        }
    }

    protected void map(LongWritable lno, Text line, Context context) throws IOException, InterruptedException {
        String attemp = context.getTaskAttemptID().toString();
        LocatedBlock lb = null;
        HdfsFileStatus hfstatus = null;
        String opath = "";
        RefreshFileSize rfs = null;
        String path=line.toString();/*输入文件里的每一行*/
        Configuration conf = context.getConfiguration();
        String nn = conf.get("fs.default.name","hdfs://192.168.60.41:8020");/*namenode*/
        String targetdir=conf.get("target.directory");
        URI uri;
        try {
            uri = new URI(nn);

        } catch (URISyntaxException e) {
            throw new  IllegalArgumentException(e);
        }
        dfs = new DFSClient(uri,conf);
        s4h = new Service4Hep(dfs,conf);
        rfs = new RefreshFileSize(dfs);
        int lastSlash=path.lastIndexOf('/');/*获取文件名*/
        String dest=targetdir+'/'+path.substring(lastSlash+1);/*完整的目标逻辑路径*/
        System.out.println(dest);
        hfstatus = s4h.createFile(dest,true,(short) 1,2147483648L);/*新建文件获取元数据信息*/
        lb = s4h.addBlock(dest,hfstatus.getFileId());/*为文件分配block*/
        String tmp =  s4h.getTmpFile(lb);
        opath = tmp+"/"+lb.getBlock().getBlockName();/*实际的物理路径*/
        System.out.println(opath);
        rfs.put(dest,hfstatus.getFileId(),opath);
        try {

            long size= client.getSize(path);
            System.out.println(size);
            totalsize+=size;
            DataSink sink =
                    new FileRandomIO(new RandomAccessFile(opath,"rw"));
            System.out.println("==================");
            long start=System.currentTimeMillis();
            client.extendedGet(path,size,sink,null);
            totaltime+=System.currentTimeMillis()-start;
            System.out.println("*************************");

        }catch (Exception e){
            e.printStackTrace();
        }

            rfs.remove(dest);
            File f = new File(opath);
            if(f.exists()){
                s4h.calculatefileCheckSums(opath,lb);
                lb.getBlock().setNumBytes(f.length());
                s4h.close(lb,dest,hfstatus.getFileId());
            }else{
                dfs.delete(dest,true);
            }
            rfs.close();
            s4h.close();
            dfs.close();
        System.out.println("!!!!!!!!!!!");
    }

}

